package com.flink.hbase.multithreaded;

import com.flink.hbase.User;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;

/**
 * 基于多线程的 User 数据源
 * 继承 HBaseMultithreadedSource，专门用于读取 User 数据
 */
public class UserMultithreadedSource extends HBaseMultithreadedSource<User> {
    private static final Logger LOG = LoggerFactory.getLogger(UserMultithreadedSource.class);
    
    private static final String COLUMN_USER_ID = "userId";
    private static final String COLUMN_NAME = "name";
    private static final String COLUMN_AGE = "age";
    private static final String COLUMN_GENDER = "gender";
    private static final String COLUMN_TIMESTAMP = "timestamp";
    private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    
    public UserMultithreadedSource(String tableName, String columnFamily, 
                                 int pageSize, int asyncThreads, 
                                 long scanTimeout, Map<String, String> hbaseConfig) {
        super(tableName, columnFamily, pageSize, asyncThreads, scanTimeout, hbaseConfig);
    }
    
    @Override
    protected User parseResult(Result result) throws Exception {
        if (result == null || result.isEmpty()) {
            return null;
        }
        
        try {
            // 从 rowkey 获取 userId
            String userId = Bytes.toString(result.getRow());
            
            // 获取列族字节数组
            byte[] cfBytes = Bytes.toBytes(getColumnFamily());
            
            // 解析各个字段
            String name = getStringValue(result, cfBytes, COLUMN_NAME);
            int age = getIntValue(result, cfBytes, COLUMN_AGE, 0);
            String gender = getStringValue(result, cfBytes, COLUMN_GENDER);
            String timestampStr = getStringValue(result, cfBytes, COLUMN_TIMESTAMP);
            
            // 解析时间戳
            LocalDateTime timestamp = LocalDateTime.now();
            if (timestampStr != null && !timestampStr.isEmpty()) {
                try {
                    timestamp = LocalDateTime.parse(timestampStr, DATETIME_FORMATTER);
                } catch (Exception e) {
                    LOG.warn("Failed to parse timestamp: {}, using current time", timestampStr);
                }
            }
            
            // 数据验证
            if (userId == null || userId.isEmpty()) {
                LOG.warn("Invalid userId in result, skipping record");
                return null;
            }
            
            if (name == null || name.trim().isEmpty()) {
                LOG.warn("Invalid name for userId: {}, skipping record", userId);
                return null;
            }
            
            if (age < 0 || age > 150) {
                LOG.warn("Invalid age {} for userId: {}, setting to 0", age, userId);
                age = 0;
            }
            
            return new User(userId, name.trim(), age, gender, timestamp);
            
        } catch (Exception e) {
            LOG.error("Error parsing HBase result", e);
            return null;
        }
    }
    
    /**
     * 获取字符串值
     */
    private String getStringValue(Result result, byte[] columnFamily, String qualifier) {
        byte[] value = result.getValue(columnFamily, Bytes.toBytes(qualifier));
        return value != null ? Bytes.toString(value) : null;
    }
    
    /**
     * 获取整数值
     */
    private int getIntValue(Result result, byte[] columnFamily, String qualifier, int defaultValue) {
        byte[] value = result.getValue(columnFamily, Bytes.toBytes(qualifier));
        if (value != null) {
            try {
                return Integer.parseInt(Bytes.toString(value));
            } catch (NumberFormatException e) {
                LOG.warn("Invalid integer value for {}: {}", qualifier, Bytes.toString(value));
            }
        }
        return defaultValue;
    }
    
    /**
     * 获取列族名称（用于子类访问）
     */
    private String getColumnFamily() {
        // 通过反射获取父类的 columnFamily 字段
        try {
            java.lang.reflect.Field field = getClass().getSuperclass().getDeclaredField("columnFamily");
            field.setAccessible(true);
            return (String) field.get(this);
        } catch (Exception e) {
            LOG.error("Failed to get column family", e);
            return "cf"; // 默认值
        }
    }
    
    /**
     * 工厂方法，创建 UserMultithreadedSource 实例
     */
    public static UserMultithreadedSource create(String tableName, String columnFamily, 
                                               int pageSize, int asyncThreads, 
                                               long scanTimeout, Map<String, String> hbaseConfig) {
        return new UserMultithreadedSource(tableName, columnFamily, pageSize, 
                                         asyncThreads, scanTimeout, hbaseConfig);
    }
} 